Documents the CometPythonMapInArrowExec optimization, including
supported APIs, configuration, usage example, and how to verify
the optimization is active in query plans.

…ions

Fix three issues that prevented test_pyarrow_udf.py from running:

1. mapInArrow callbacks must accept Iterator[pa.RecordBatch] and yield
   batches. The previous single-batch signatures crashed with
   "'map' object has no attribute 'to_pandas'".
2. PySpark DataFrame has no `queryExecution` attribute. Use
   `_jdf.queryExecution().executedPlan().toString()` instead.
3. Replace soft plan-string heuristics with assertions that fail loudly
   if the optimization regresses. Match on `CometPythonMapInArrow` (no
   `Exec` suffix in the plan toString) and assert no `ColumnarToRow`
   transition is present.

- Rewrite test_pyarrow_udf.py as a pytest module. A session-scoped
  SparkSession fixture builds the Comet-enabled session once and a
  parametrized `accelerated` fixture toggles
  spark.comet.exec.pythonMapInArrow.enabled per test, so each case runs
  under both the optimized and fallback paths and asserts the expected
  plan operator (`CometPythonMapInArrow` vs vanilla `PythonMapInArrow`).
  The jar is auto-discovered from spark/target by matching the installed
  pyspark version, or taken from the COMET_JAR env var.
- Add a dedicated `PyArrow UDF Tests` workflow that builds Comet against
  Spark 3.5 / Scala 2.12, installs pyspark/pyarrow/pandas/pytest, and
  runs the new pytest module.
- Add CometPythonMapInArrowSuite to the `exec` suite list in both
  pr_build_linux.yml and pr_build_macos.yml so the JVM-side suite is
  exercised on every PR.

Replace the narrow paths allowlist with the same paths-ignore list used
by pr_build_linux.yml so the workflow runs on any source change that
could affect Comet's PyArrow UDF execution path, not just the few files
explicitly named.

The PR's `CometPythonMapInArrowExec` and `EliminateRedundantTransitions`
rule directly reference Spark 3.5 APIs that differ across supported
Spark versions: the `ArrowPythonRunner` constructor (4 distinct
signatures across 3.4/3.5/4.0/4.1+/4.2), `arrowUseLargeVarTypes`,
`JobArtifactSet`, `MapInBatchExec.isBarrier`, and the `PythonMapInArrowExec`
type itself (renamed to `MapInArrowExec` in 4.0+). This breaks compile
on every profile other than 3.5.

Introduce a per-version `ShimCometPythonMapInArrow` trait under
`org.apache.spark.sql.comet.shims` (placed in the spark namespace so
it can reach `private[spark]` members) that:

* matches the Spark-version-specific MapInArrow / MapInPandas exec types
  and exposes their `(func, output, child, isBarrier, evalType)` tuple,
* constructs the right `ArrowPythonRunner` for the version,
* hides `arrowUseLargeVarTypes` / `JobArtifactSet` / `getPythonRunnerConfMap`
  behind helper methods.

Spark 3.4 lacks the prerequisite APIs (no `isBarrier`, no `JobArtifactSet`,
no `arrowUseLargeVarTypes`), so its shim returns `None` from the matchers
and the optimization is a no-op there.

The default `amd64/rust` image is Debian 13 (trixie), where the system
`python3` is 3.13 and there is no `python3.11` apt package. The workflow
installed `python3.11` explicitly, which fails on trixie with `Unable to
locate package python3.11`.

Switching to `rust:bookworm` gives a Debian 12 base where `python3` is
3.11, matching the job name and pyspark 3.5.x's supported runtime.

Spark launches Python workers in fresh subprocesses that look up python3
on PATH. Without PYSPARK_PYTHON, workers use the system python (no pyarrow
installed) and UDF execution fails with ModuleNotFoundError. Point both
PYSPARK_PYTHON and PYSPARK_DRIVER_PYTHON at /tmp/venv/bin/python so workers
inherit the same interpreter that pytest uses.

Add coverage for cases that the original pytest module did not exercise:

- mapInPandas (claimed supported, previously zero coverage)
- Null preservation across long and string columns via Arrow passthrough
- Empty input from a CometScan via filter pushdown
- Python exception propagation (sentinel must surface in driver-side error)
- DecimalType(18,6), DateType, TimestampType round-trip with nulls
- ArrayType<Int> and nested StructType, including null arrays/structs and
  arrays containing null elements
- repartition between scan and UDF (correctness only; the optimization
  itself does not fire across a vanilla Exchange and is documented as
  such in the test)

Generalize _assert_plan_matches_mode to take the vanilla node name so the
fallback assertion can match either PythonMapInArrow or MapInPandas.

Expand the user guide with the limitations a user should know before
enabling the experimental optimization:

- The remaining row-to-Arrow round-trip inside the Python runner is
  documented more precisely (the input goes through ColumnarBatch.rowIterator
  to feed ArrowPythonRunner, which re-encodes to Arrow IPC).
- A vanilla Spark Exchange between the Comet scan and the UDF prevents
  the optimization from firing. Users must configure Comet's native
  shuffle manager at session startup to keep the data columnar.
- Spark 3.4 lacks the prerequisite APIs and the feature is a no-op there.
- isBarrier is captured by the operator constructor but not yet
  propagated to the Python runner.

Also explain the AQE display quirk: with AQE on and a shuffle present,
the pre-execution plan shows the unoptimized form because the rule
only sees the materialized subplan after stage execution. Running an
action and re-inspecting explain() reveals the optimized plan.

Standalone Python script that times df.mapInArrow(passthrough).count()
and the equivalent mapInPandas query with the optimization toggled on
and off. Numbers are wall-clock seconds, so they include the Python
worker, Arrow IPC, and downstream count() costs. That is the right
unit for a feature whose user surface is Python: it shows what
fraction of end-to-end time the optimization shaves off, not just the
JVM-side delta in isolation.

Three workloads exercise the dimension where the optimization helps
most:

- narrow primitives (long, int, double)
- mixed with strings (variable-length encoding)
- wide rows (50 columns, projection cost scales with column count)

Local smoke run with 200k rows shows 1.17x to 1.45x speedup across
mapInArrow and mapInPandas, narrow/wide schemas. The script is
configurable via BENCHMARK_ROWS / BENCHMARK_WARMUP / BENCHMARK_ITERS
env vars for users who want longer or shorter runs.

…p ci]

…[skip ci]

…ck [skip ci]

…ddress [skip ci]

Adds a helper class in comet-common that copies Arrow buffer bytes from a
CometDecodedVector to caller-supplied memory addresses without exposing shaded
Arrow types across the module boundary. Uses getFieldBuffers()/getChildrenFromFields()
traversal consistent with VectorUnloader so buffer ordering matches between source
(shaded) and destination (unshaded) sides.

…[skip ci]

Rewrites CometColumnarPythonInput to copy Comet vector bytes via
CometVectorIpcCopier (long-address API) rather than casting shaded FieldVector
to unshaded FieldVector, which caused ClassCastException at runtime.

Additional fixes for correct Arrow IPC semantics:
- Fill struct validity buffer with 0xFF so Python sees non-null struct rows
- Set lastSet before setValueCount on variable-width and list vectors to prevent
  fillHoles from overwriting correctly copied offset buffers
- Process nodes bottom-up so parent setValueCount cascade does not clobber
  children that have not yet had lastSet updated

…ndary

# Conflicts:
#	spark/pom.xml
#	spark/src/main/java/org/apache/comet/vector/CometVectorIpcCopier.java

The long[]-address indirection through CometVectorIpcCopier existed because
comet-common shaded org.apache.arrow.* into org.apache.comet.shaded.arrow.*,
making source vectors and Spark's IPC root different JVM types. After apache#4325
moved most JVM code into comet-spark and dropped the shading, both sides see
the same Arrow classes — the helper is no longer needed.

Replace with a direct walk of the source/destination FieldVector trees using
ArrowBuf.setBytes for the buffer copy. Same per-buffer memcpy semantics; the
cross-RootAllocator constraint that blocks true zero-copy is independent of
shading and still tracked in apache#4294.

Adds pytest cases for the data-type branches in CometColumnarPythonInput
that were previously unexercised: numeric scalars (boolean/byte/short/float),
binary, timestamp NTZ, map, and a deeply nested array/struct combination.

Falls back to vanilla Spark when spark.sql.execution.arrow.useLargeVarTypes
is enabled. With that conf on, Spark widens StringType/BinaryType to
8-byte-offset variants in the destination IPC root while Comet's source
vectors keep 4-byte offsets, so the per-buffer memcpy in copyVector would
corrupt the offset buffer.

While narrowing the rule to gate on largeVarTypes, also fix a pre-existing
greedy match: the MapInBatch case used `p: SparkPlan` with the pyarrow conf
as a guard, which matched every plan when the conf was on and consumed the
later CometShuffleExchangeExec arm. The case now gates on a structural check
via eligibleMapInBatchInfo so unrelated plans flow through.

- Move the 3.4 / 3.5 ShimCometMapInBatch stubs into a single spark-3.x shim
  (they were byte-identical). The matchers still return None on both versions
  so the rule is a no-op on Spark 3.x.

- Replace the eligibleMapInBatchInfo guard + .get unpack with an
  EligibleMapInBatch extractor that runs the matchers and conf reads once
  per visited plan.

- Add arrowUseLargeVarTypes(conf) to ShimSQLConf so the rule no longer reads
  the conf stringly. 4.x and 3.5 forward to the typed accessor; 3.4 falls
  back to getConfString because that version has no accessor.

- Hoist the per-batch VectorUnloader in CometColumnarPythonInput to a lazy
  val. getRecordBatch reads root.getFieldVectors on every call so reuse is
  safe; this drops one allocation per batch.

- Clarify the comment on cometCodec: 4.0.x has no
  SQLConf.arrowCompressionCodec accessor (added after 4.0 branch was cut),
  so a typed ShimSQLConf forwarder would still need a stringly fallback for
  the 4.0 build. The 4.1+ codec instances live in the separate
  arrow-compression artifact, which Comet does not depend on; the
  CompressionCodec.Factory path keeps that dependency contained.

Addresses mbutrovich's items 5, 6, 8, 10 on apache#4234.

The PR description, CometColumnarPythonInput header, and pyarrow-udfs.md all
blamed the per-buffer copy on 'Comet's Parquet readers each constructing
their own RootAllocator'. The repo only has one process-wide RootAllocator
(CometArrowAllocator), and native scan does Parquet reading on the Rust
side: arrow buffers cross the boundary via Arrow C Data Interface, not a
JVM allocator.

The actual blocker on TransferPair is that imported buffers carry a
ReferenceManager whose release routes through FFI, while Spark's
destination IPC root is a child of ArrowUtils.rootAllocator. The two
reference managers cannot share buffers.

Reframe the per-batch work as 'two copies, one structural':
- copy 1 (Comet -> destination IPC root) is droppable, tracked in apache#4294
- copy 2 (root -> pipe via VectorUnloader / MessageSerializer) is the
  structural floor; Spark's transport to Python is fork + pipe + Arrow
  IPC, so the bytes must reach the pipe at least once

Addresses mbutrovich's items 1 and 3 (framing) on apache#4234. The PR
description update is a separate step.

Hand-written cases that pin the boundaries mbutrovich called out as gaps:

- decimal precision sweep (1, 9, 17, 18, 19, 28, 38; scale 0/half/max)
  covering the short-decimal (long-backed) and long-decimal (16-byte
  FixedSizeBinary) paths and the 18/19 boundary
- null density sweep (0, 0.01, 0.5, 0.99, 1.0) for validity-buffer memcpy
- multi-batch per partition (batch size 16, 4000 rows in 1 partition) so
  the persistent destination IPC root is exercised across many batches
- wide schema (50 cols, mixed primitives + strings + booleans) for the
  flattened-tree address arithmetic
- mid-stream zero-row batch so setValueCount(0) + validity sizing is
  hit while the iterator continues
- transforming array UDF (reverse each list) to catch symmetric
  encode/decode mistakes that a passthrough would invert

A randomised fuzz harness (analogous to CometCodegenFuzzSuite) is the
right next step for the recursive vector-tree walk; deferred to a
separate follow-on.

CometSparkSessionExtensions.isCometLoaded short-circuits the whole
extension (returning false; no rules registered) when
spark.comet.exec.shuffle.enabled is true but spark.shuffle.manager is
not Comet's manager. The pytest conftest only sets the basic Comet
configs, so this guard fired and CometScanRule never ran. The plan
stayed vanilla Parquet, the rewrite chain never had a Comet columnar
producer to match, and every [accelerated] assertion that checks for
CometMapInBatch failed.

These tests do not exercise shuffle, so disable Comet shuffle in the
session. Comet's scan and exec rules then run normally and the rewrite
fires.

Diagnoses the wholesale PyArrow UDF Spark 4.0 CI failure on apache#4234.